#include <src/pcps.h>
#include <src/disk_io.h>

#include <fstream>

//--------------------------------------------------------------------------------
// pcps::disk_io_detail -- function used by the disk io thread
//--------------------------------------------------------------------------------

void * pcps::disk_io_detail(void * arg) {

  // get a pointer to the disk io manager
  pcps::Disk_IO_Manager * const manager = (pcps::Disk_IO_Manager *)arg;

  // perform disk input/output operations until instructed to terminate
  while (true) {

    // lock the manager's mutex
    pthread_mutex_lock(&manager->_cond_mutex);

    // if instructed to, exit the loop and terminate the thread
    if (manager->_terminate) {

      // unlock the manager's mutex and exit the loop
      pthread_mutex_unlock(&manager->_cond_mutex);
      break;

    // complete the next task in the queues
    } else if (!manager->_type_queue.empty()) {

      // get instructions from the queues
      const char job                = manager->_type_queue.front();
      const size_t n                = manager->_size_queue.front();
      char * const data             = manager->_data_queue.front();
      bool * const flag             = manager->_flag_queue.front();
      pthread_mutex_t * const mutex = manager->_mutx_queue.front();
      pthread_cond_t * const cond   = manager->_cond_queue.front();
      std::string * const filename  = manager->_name_queue.front();

      // remove task from queues
      manager->_type_queue.pop();
      manager->_size_queue.pop();
      manager->_data_queue.pop();
      manager->_flag_queue.pop();
      manager->_mutx_queue.pop();
      manager->_cond_queue.pop();
      manager->_name_queue.pop();

      // if we are writing to memory and the memory is not ready, allocate the memory
      if ( job == 'w' && !manager->_use_disk ) {
        if ( manager->_virtual_disk_sizes.count(*filename) == 0 ) {
          if ( MPI::COMM_WORLD.Get_rank() == 0 )
            std::cout << boost::format("creating %i bytes of virtual disk space for file \"%s\"") % n % filename->c_str() << std::endl;
          manager->_virtual_disk_sizes[*filename] = n;
          manager->_virtual_disk[*filename] = pcps::allocate_shared_array<char>(n);
        }
        if ( manager->_virtual_disk_sizes[*filename] < n ) {
          if ( MPI::COMM_WORLD.Get_rank() == 0 )
            std::cout << boost::format("enlarging virtual disk space for file \"%s\" to %i bytes") % filename->c_str() % n << std::endl;
          manager->_virtual_disk_sizes[*filename] = n;
          manager->_virtual_disk[*filename] = boost::shared_array<char>();          // deallocate
          manager->_virtual_disk[*filename] = pcps::allocate_shared_array<char>(n); // reallocate
        }
      }

      // get memory location if we are reading from or writing to disk
      char * mem_loc;
      if ( !manager->_use_disk ) {
        if ( manager->_virtual_disk.count(*filename) == 0 )
          throw pcps::Exception( (boost::format("pcps::disk_io_detail failed to find the file \"%s\" on the virtual disk")
                                  % filename->c_str()).str() );
        mem_loc = manager->_virtual_disk[*filename].get();
      }

      // unlock the manager's mutex so that other threads may schedule more jobs while the task is in progress
      pthread_mutex_unlock(&manager->_cond_mutex);

      // lock mutex corresponding to this task so that other threads will know it is not finished
      pthread_mutex_lock(mutex);

      // perform a write task
      if (job == 'w') {

        // write to disk
        if (manager->_use_disk) {

          // open the binary file for writing
          std::ofstream f(filename->c_str(), std::ios::out | std::ios::binary);

          // check that the file opened successfully
          if (!f.is_open())
            throw pcps::Exception( (boost::format("pcps::disk_io_detail failed to open file \"%s\" for writing")
                                    % filename->c_str()).str() );

          // write the data
          f.write(data, n);

          // check that the write was successful
          if (f.bad())
            throw pcps::Exception( (boost::format("pcps::disk_io_detail failed to write %i bytes to file \"%s\"")
                                    % n % filename->c_str()).str() );

          // close the file
          f.close();

        // write to memory
        } else {
          std::memcpy(mem_loc, data, n);
        }

      // perform a read task
      } else if (job == 'r') {

        // read from disk
        if (manager->_use_disk) {

          // open the binary file for reading
          std::ifstream f(filename->c_str(), std::ios::in | std::ios::binary);

          // check that the file opened successfully
          if (!f.is_open())
            throw pcps::Exception( (boost::format("pcps::disk_io_detail failed to open file \"%s\" for reading")
                                    % filename->c_str()).str() );

          // read the data
          f.read(data, n);

          // check that the read was successful
          if (f.fail() || f.gcount() != n)
            throw pcps::Exception( (boost::format("pcps::disk_io_detail failed to read %i bytes from file \"%s\"")
                                    % n % filename->c_str()).str() );

          // close the file
          f.close();

        // read from memory
        } else {
          std::memcpy(data, mem_loc, n);
        }

      // uknown job type
      } else {
        throw pcps::Exception( (boost::format("pcps::disk_io_detail accepts only 'r' and 'w' as job type characters, you requested '%s'")
                                % job).str() );
      }

      // set the flag to true so that other threads know the task is finished
      *flag = true;

      // signal any threads that are waiting for the task to finish
      pthread_cond_broadcast(cond);

      // unlock mutex corresponding to this task
      pthread_mutex_unlock(mutex);

    // if there are no tasks in the queues, wait for more work
    } else {
      pthread_cond_wait(&manager->_cond, &manager->_cond_mutex);
      pthread_mutex_unlock(&manager->_cond_mutex);
    }

  }

}
